#include <sys/types.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <netdb.h>
#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <string.h>
#include <unistd.h>

#include "config.h"
#include "libxsp.h"

#include "psd_conn.h"
#include "psd_modules.h"
#include "psd_measurement.h"
#include "psd_conn_tcp.h"
#include "psd_conn_file.h"
#include "psd_config.h"
#include "psd_tpool.h"
#include "psd_logger.h"

#include "compat.h"

void *psd_lsink_accept_handler(void *arg);
void *lsink_handle_client(void *arg);
void fill_filename(char *filename, int filename_len, const char *conn_id);
int psd_lsink_init();
void psd_lsink_read_config();

static int *sockets;
static int num_sockets;

static psdModule psd_lsink_module = {
	.desc = "XSP Depot Sink Module",
	.dependencies = "tcp",
	.init = psd_lsink_init
};

struct psd_lsink_config_t {
	int port;
	int recv_bufsize;
	int recv_timeout;
	char *filename_pattern;
	char *output_dir;
};

static struct psd_lsink_config_t psdLsinkConfig = {
	.port =         5003,
	.recv_bufsize = 0,
	.recv_timeout = 0,
	.filename_pattern  = "psd_%d_%t_output",
	.output_dir = "/tmp"
};

void psd_lsink_read_config() {
	int val;
	char *str_val;

	if (psd_depot_settings_get("lsink", "filename_pattern", &str_val) == 0) {
		psdLsinkConfig.filename_pattern = str_val;
	}

	if (psd_depot_settings_get("lsink", "output_dir", &str_val) == 0) {
		psdLsinkConfig.output_dir = str_val;
	}

	if (psd_depot_settings_get_int("lsink", "port", &val) == 0) {
		psdLsinkConfig.port = val;
	}

	if (psd_depot_settings_get_int("lsink", "recv_bufsize", &val) == 0) {
		psdLsinkConfig.recv_bufsize = val;
	}

	if (psd_depot_settings_get_int("lsink", "recv_timeout", &val) == 0) {
		psdLsinkConfig.recv_timeout = val;
	}
}

psdModule *module_info() {
	return &psd_lsink_module;
}

int psd_lsink_init() {
	struct addrinfo *ai;
	int i;

	psd_lsink_read_config();

	sockets = listen_port(IPPROTO_TCP, PF_UNSPEC, psdLsinkConfig.port, &num_sockets, &ai);
	if (!sockets) {
		psd_err(0, "couldn't listen on port: %d", psdLsinkConfig.port);
		goto error_exit;
	}

	freeaddrinfo(ai);

	for(i = 0; i < num_sockets; i++) {

		if (psdLsinkConfig.recv_bufsize > 0) {
			setsockopt(sockets[i], SOL_SOCKET, SO_RCVBUF,
					&(psdLsinkConfig.recv_bufsize),
					sizeof(int));
		}
	}

	if (psd_tpool_exec(psd_lsink_accept_handler, NULL) != 0) {
		psd_err(0, "couldn't execute socket handler thread");
		goto error_exit2;
	}

	return 0;

error_exit2:
	for(i = 0; i < num_sockets; i++)
		close(sockets[i]);
error_exit:
	return -1;
}

void *psd_lsink_accept_handler(void *arg) {
	SOCKLEN_T peer_sa_len;
	fd_set socket_set;
	int n, i, high;
	struct sockaddr_storage sa;
	int cfd;

	while(1) {

		FD_ZERO(&socket_set);

		high = 0;

		for(i = 0; i < num_sockets; i++) {
			FD_SET(sockets[i], &socket_set);
			if (sockets[i] > high)
				high = sockets[i];
		}

		n = select(high + 1, &socket_set, NULL, NULL, NULL);

		for(i = 0; i < num_sockets; i++) {
			if (FD_ISSET(sockets[i], &socket_set)) {
				psdConn *conn;

				peer_sa_len = sizeof(struct sockaddr_storage);

				cfd = accept(sockets[i], (struct sockaddr *) &sa, &peer_sa_len);
				if (cfd < 0) {
					continue;
				}

				conn = psd_conn_tcp_alloc(cfd, 0);
				if (!conn) {
					close(cfd);
					continue;
				}

				psd_tpool_exec(lsink_handle_client, conn);
			}
		}
	}

	return NULL;
}

void* lsink_handle_client(void *arg) {
	int file_fd;
	uint64_t amt;
	char *buf;
	int bufsize;
	psdConn *conn = arg;
	psdConn *file_conn;
	char filename[1024];

	psd_info(1, "received connection from %s", conn->description);

	if (psdLsinkConfig.recv_bufsize == 0) {
		bufsize = 8*1024*1024;
	} else {
		bufsize = psdLsinkConfig.recv_bufsize;
	}

	psd_info(5, "buffer size set to %d", bufsize);

	buf = malloc(sizeof(char) * bufsize);
	if (!buf) {
		psd_err(1, "failed to allocate bufferspace for %s", conn->description);
		goto error_exit;
	}

	fill_filename(filename, sizeof(filename), conn->description);

	psd_info(1, "writing data from %s to %s", conn->description, filename);

	file_fd = open(filename, O_WRONLY | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR);
	if (file_fd < 0) {
		psd_err(1, "failed to open %s", filename);
		goto error_exit2;
	}

	file_conn = psd_alloc_conn_file(file_fd);
	if (!file_conn) {
		psd_err(1, "failed to allocate file connection for %s", conn->description);
		goto error_exit3;
	}

	amt = 0;

	while(1) {
		int n;
		n = psd_conn_read(conn, buf, bufsize, 0);
		if (n <= 0)
			goto out;

		psd_info(10, "received %d bytes from %s", n, conn->description);

		psd_conn_write(file_conn, buf, n, 0);

		amt += n;
	}
out:
	psd_info(1, "Connection to %s closed. Wrote %llu bytes to %s",
			conn->description, amt, filename);

	free(buf);
	psd_conn_shutdown(file_conn, PSD_RECV_SIDE | PSD_SEND_SIDE);
	psd_conn_free(file_conn);
	psd_conn_shutdown(conn, PSD_RECV_SIDE | PSD_SEND_SIDE);
	psd_conn_free(conn);

	return NULL;

error_exit3:
	close(file_fd);
error_exit2:
	free(buf);
error_exit:
	psd_conn_shutdown(conn, PSD_RECV_SIDE | PSD_SEND_SIDE);
	psd_conn_free(conn);
	return NULL;
}

void fill_filename(char *filename, int filename_len, const char *conn_id) {
	char *c1, *c2;
	int i;

	c1 = psdLsinkConfig.filename_pattern;
	c2 = filename;

	strlcpy(filename, psdLsinkConfig.output_dir, filename_len);
	strlcat(filename, "/", filename_len);
	c2 = filename + strlen(filename);

	while(*c1 != '\0') {
		switch(*c1) {
			case '/':
				*c2 = '_';
				break;
			case '%':
				if (*(c1 + 1) == 'd') {
					strlcpy(c2, conn_id, filename_len);
					c1++;
					for(i = 0; i < strlen(conn_id); i++) {
						if (c2[i] == '/')
							c2[i] = '_';
					}
					c2 += strlen(conn_id) - 1;
				} else if (*(c1 + 1) == 't') {
					char time_buf[10];
					sprintf(time_buf, "%d", (int) time(NULL));
					c1++;
					strlcpy(c2, time_buf, filename_len);
					c2 += strlen(time_buf) - 1;
				} else {
					*c2 = '%';
				}
				break;
			default:
				*c2 = *c1;
				break;
		}

		c1++;
		c2++;
	}
}
